package com.atbeijing.D05;


import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 模拟滚动(增量)聚合 求sensor_1的平均温度
 */
public class Example8 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .filter(r -> r.id.equals("sensor_1"))
                .keyBy(r -> r.id)
                .process(new MyKeyedProcessFunction3())
                .print();

        env.execute();
    }

    public static class MyKeyedProcessFunction3 extends KeyedProcessFunction<String, SensorReading,String>{
        //tuple<温度,数据个数>
        private ValueState<Tuple2<Double,Long>> myState;
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            myState=getRuntimeContext().getState(new ValueStateDescriptor<Tuple2<Double, Long>>("my-state", Types.TUPLE(Types.DOUBLE,Types.LONG)));
        }

        @Override
        public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {
            if (myState.value()==null){
                //第一条数据
                myState.update(Tuple2.of(value.temperature,1L));
            }else {
                Tuple2<Double, Long> t2 = myState.value();
                //温度累加,个数+1
                myState.update(Tuple2.of(t2.f0+value.temperature,t2.f1+1L));
            }
            //输出平均值
            Tuple2<Double, Long> t22 = myState.value();
            out.collect("平均温度为:\t"+t22.f0/t22.f1);
        }


    }

}
